package com.zhb.cloud.curator;

import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.apache.commons.io.Charsets;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorCrudTest {
	/**
	 * NAME_SPACE
	 */
	private static final String NAME_SPACE = "zhb/zk";
	/**
	 * PATH
	 */
	private static final String PATH = "/zhb/test";
	/**
	 * zookeeper hosts address
	 */
	private static final String CONNECT_STRING = "192.168.1.103:2181";
	/**
	 * session timeout in milliseconds
	 */
	public static final int ZOOKEEPER_SESSION_TIMEOUT = 10 * 1000;

	/**
	 * connection timeout in milliseconds
	 */
	public static final int ZOOKEEPER_CONNECTION_TIMEOUT = 10 * 1000;

	/**
	 * max number of times to retry
	 */
	public static final int ZOOKEEPER_RETRY_TIMES = 3;

	/**
	 * initial amount of time to wait between retries in milliseconds
	 */
	public static final int ZOOKEEPER_RETRY_INTERVAL = 1000;

	public static void main(String[] args) throws Exception {
		// 获得zk客户端
		CuratorFramework curator = getCuratorFramework(CONNECT_STRING, NAME_SPACE);
		ExecutorService executor = Executors.newCachedThreadPool();
		curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
				.inBackground(new BackgroundCallback() {
					public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
						System.out.println("Code：" + event.getResultCode());
						System.out.println("Type：" + event.getType());
						System.out.println("Path：" + event.getPath());
					}
				}, executor).forPath(PATH, "path下的原始内容".getBytes());

		Thread.sleep(5000); // 为了能够看到回调信息
		String data = new String(curator.getData().forPath(PATH)); // 获取节点数据
		System.out.println(data);
		Stat stat = curator.checkExists().forPath(PATH); // 判断指定节点是否存在
		System.out.println(stat);
		curator.setData().forPath(PATH, "path下的新内容".getBytes()); // 更新节点数据
		data = new String(curator.getData().forPath(PATH));
		System.out.println(data);
		List<String> children = curator.getChildren().forPath("/zhb"); // 获取子节点
		for (String child : children) {
			System.out.println(child);
		}
		// 放心的删除节点，deletingChildrenIfNeeded()方法表示如果存在子节点的话，同时删除子节点
		curator.delete().guaranteed().deletingChildrenIfNeeded().forPath("/zhb");
		curator.close();
	}

	/***
	 * @Title: getCuratorFramework
	 * @author: zhb
	 * @date: 2017年11月5日 上午10:02:49
	 * @Description: 获取zk客户端
	 * @param connectString
	 * @param namespace
	 * @return
	 */
	public static CuratorFramework getCuratorFramework(String connectString, String namespace) {
		if (StringUtils.isEmpty(connectString)) {
			throw new NullPointerException("empty zookeeper connect string");
		}
		// normalize namespace
		if (!StringUtils.isBlank(namespace)) {
			namespace = namespace.replaceAll("^/+", "");
		}
		// retry policy

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(ZOOKEEPER_RETRY_INTERVAL, ZOOKEEPER_RETRY_TIMES);

		// default data: host address
		String host = getHostAddress();
		byte[] defaultData = null;
		if (host != null) {
			defaultData = host.getBytes(Charsets.toCharset("UTF-8"));
		}
		// build zk client
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(connectString)
				.sessionTimeoutMs(ZOOKEEPER_SESSION_TIMEOUT).connectionTimeoutMs(ZOOKEEPER_CONNECTION_TIMEOUT)
				.canBeReadOnly(true).retryPolicy(retryPolicy).defaultData(defaultData);
		if (!StringUtils.isBlank(namespace)) {
			builder.namespace(namespace);
		}
		CuratorFramework client = builder.build();
		// start zk client
		if (client.getState() != CuratorFrameworkState.STARTED) {
			client.start();
		}
		return client;
	}

	/**
	 * Returns localHost's IP address.
	 * 
	 * @return
	 */
	public static String getHostAddress() {
		String ip = null;
		try {
			// InetAddress
			ip = InetAddress.getLocalHost().getHostAddress();

			// NetworkInterface address
			List<String> list = new ArrayList<String>();
			Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
			while (interfaces.hasMoreElements()) {
				NetworkInterface iface = interfaces.nextElement();
				// filters out 127.0.0.1 and inactive interfaces
				if (iface.isLoopback() || !iface.isUp())
					continue;

				Enumeration<InetAddress> addresses = iface.getInetAddresses();
				while (addresses.hasMoreElements()) {
					InetAddress addr = addresses.nextElement();
					list.add(addr.getHostAddress());
				}
			}

			// sort
			Collections.sort(list);
			if (!list.isEmpty()) {
				ip = list.get(0).toString();
			}
		} catch (SocketException | UnknownHostException e) {
		}
		return ip;
	}

}
